package com.starrocks.service;

import com.alibaba.fastjson2.JSON;
import com.starrocks.adapter.SyncConfigAdapter;
import com.starrocks.config.SyncConfig;
import com.starrocks.entity.SyncConfigEntity;
import com.starrocks.entity.SyncTableConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于时间戳的增量同步服务
 * 
 * 说明：由于StarRocks的binlog协议与MySQL不兼容，mysql-binlog-connector-java库无法使用。
 * 因此采用基于update_time字段的增量同步方案。
 * 
 * 优点：
 * - 实现简单，稳定可靠
 * - 不依赖binlog协议
 * - 性能可控
 * 
 * 缺点：
 * - 需要表有update_time字段
 * - 轮询机制，有一定延迟（默认5秒）
 * - 无法捕获DELETE操作（除非使用软删除）
 */
@Slf4j
@Service
public class IncrementalSyncService {

    @Autowired
    @Qualifier("starrocksJdbcTemplate")
    private JdbcTemplate starrocksJdbcTemplate;

    @Autowired
    private SyncConfig syncConfig;

    @Autowired
    private DataSyncHandler dataSyncHandler;
    
    @Autowired(required = false)
    private SyncConfigService syncConfigService;
    
    @Autowired(required = false)
    private SyncLogService syncLogService;
    
    /**
     * 全量同步线程池
     */
    @Autowired
    @Qualifier("fullSyncExecutor")
    private ThreadPoolTaskExecutor fullSyncExecutor;

    // 记录每个表的最后同步时间
    private final Map<String, LocalDateTime> lastSyncTimeMap = new ConcurrentHashMap<>();
    
    // 当前生效的同步配置列表（从数据库加载或从YAML加载）
    private List<SyncTableConfig> activeTableConfigs = new ArrayList<>();
    
    // 标记哪些表正在进行全量同步（用于跳过增量同步）
    private final Map<String, Boolean> fullSyncingMap = new ConcurrentHashMap<>();

    private volatile boolean running = false;

    /**
     * 启动增量同步
     */
    public void startSync() {
        if (running) {
            log.warn("增量同步已经在运行中");
            return;
        }

        // 加载配置（优先从数据库，其次从YAML）
        loadActiveConfigs();
        
        if (activeTableConfigs.isEmpty()) {
            log.warn("没有启用的同步配置，同步服务未启动");
            return;
        }

        log.info("启动基于时间戳的增量同步服务，监听表配置数量：{}", activeTableConfigs.size());

        // 🔥🔥🔥 第一步：先标记所有需要全量同步的表（在启动增量服务前）
        // 这样可以避免增量任务在全量同步启动前就开始执行
        for (SyncTableConfig tableConfig : activeTableConfigs) {
            if (tableConfig.getEnabled() && Boolean.TRUE.equals(tableConfig.getFullSyncOnStartup())) {
                String syncKey = getSyncKey(tableConfig);
                fullSyncingMap.put(syncKey, true);
                log.info("🔒 表 {} -> {} 标记为全量同步中，增量同步将被暂停", 
                    tableConfig.getSourceTableName(), tableConfig.getTargetTableName());
            }
        }

        // 🔥🔥🔥 第二步：启动增量同步服务（此时全量同步标记已设置）
        running = true;
        log.info("增量同步服务启动成功");

        // 🔥🔥🔥 第三步：启动全量同步任务和初始化其他表
        for (SyncTableConfig tableConfig : activeTableConfigs) {
            if (tableConfig.getEnabled()) {
                String syncKey = getSyncKey(tableConfig);
                
                // 检查是否需要全量同步
                if (Boolean.TRUE.equals(tableConfig.getFullSyncOnStartup())) {
                    log.info("⏳ 表 {} -> {} 提交到线程池异步执行全量同步...", 
                        tableConfig.getSourceTableName(), tableConfig.getTargetTableName());
                    
                    // 使用线程池异步执行全量同步，不阻塞启动流程
                    fullSyncExecutor.submit(() -> {
                        try {
                            log.info("🚀 表 {} -> {} 开始执行全量同步（异步线程）", 
                                tableConfig.getSourceTableName(), tableConfig.getTargetTableName());
                            performFullSync(tableConfig);
                        } catch (Exception e) {
                            log.error("❌ 表 {} -> {} 全量同步异常", 
                                tableConfig.getSourceTableName(), tableConfig.getTargetTableName(), e);
                        } finally {
                            // 全量同步完成，清除标记
                            fullSyncingMap.remove(syncKey);
                            log.info("✅ 表 {} -> {} 全量同步完成，已清除全量同步标记，恢复增量同步", 
                                tableConfig.getSourceTableName(), tableConfig.getTargetTableName());
                            log.info("📊 当前 fullSyncingMap 状态: {}", fullSyncingMap);
                        }
                    });
                    
                } else {
                    // 从当前时间开始同步
                    lastSyncTimeMap.put(syncKey, LocalDateTime.now());
                    log.info("表 {} -> {} 的初始同步时间点：{}", 
                        tableConfig.getSourceTableName(), 
                        tableConfig.getTargetTableName(),
                        lastSyncTimeMap.get(syncKey));
                }
            }
        }
    }
    
    /**
     * 加载配置
     * 优先从数据库加载，如果数据库没有配置则从YAML加载
     */
    private void loadActiveConfigs() {
        activeTableConfigs.clear();
        
        // 尝试从数据库加载
        if (syncConfigService != null) {
            try {
                List<SyncConfigEntity> dbConfigs = syncConfigService.listEnabledConfigs();
                if (dbConfigs != null && !dbConfigs.isEmpty()) {
                    activeTableConfigs = SyncConfigAdapter.convertToSyncTableConfigs(dbConfigs);
                    log.info("从数据库加载了 {} 个同步配置，生成 {} 个表映射关系", 
                        dbConfigs.size(), activeTableConfigs.size());
                    return;
                }
            } catch (Exception e) {
                log.error("从数据库加载配置失败，将尝试使用YAML配置", e);
            }
        }
        
        // 如果数据库没有配置，使用YAML配置
        if (syncConfig != null && syncConfig.getEnabled() && syncConfig.getTables() != null) {
            activeTableConfigs = new ArrayList<>(syncConfig.getTables());
            log.info("从YAML配置文件加载了 {} 个同步配置", activeTableConfigs.size());
        } else {
            log.warn("YAML配置未启用或没有配置表");
        }
    }
    
    /**
     * 重新加载配置（热加载）
     */
    public void reloadConfigs() {
        log.info("重新加载同步配置...");
        loadActiveConfigs();
        log.info("配置重新加载完成，当前生效配置数量：{}", activeTableConfigs.size());
    }
    
    /**
     * 生成同步键（源表+目标表）
     * 用于一对多场景，区分同一源表到不同目标表的同步进度
     */
    private String getSyncKey(SyncTableConfig tableConfig) {
        return tableConfig.getSourceTableName() + "->" + tableConfig.getTargetTableName();
    }

    /**
     * 停止增量同步
     */
    public void stopSync() {
        if (!running) {
            log.warn("增量同步未在运行");
            return;
        }

        log.info("停止增量同步服务");
        running = false;
        log.info("增量同步服务已停止");
    }

    /**
     * 定时同步任务（每5秒执行一次）
     */
    @Scheduled(fixedDelay = 5000, initialDelay = 5000)
    public void syncData() {
        if (!running) {
            return;
        }

        // 定时任务静默执行，仅在有数据时输出日志

        for (SyncTableConfig tableConfig : activeTableConfigs) {
            if (!tableConfig.getEnabled()) {
                continue;
            }

            try {
                syncTableData(tableConfig);
            } catch (Exception e) {
                log.error("同步表 {} -> {} 失败", 
                    tableConfig.getSourceTableName(), 
                    tableConfig.getTargetTableName(), e);
            }
        }
    }

    /**
     * 全量同步（优化版：使用游标分页）
     */
    private void performFullSync(SyncTableConfig tableConfig) {
        String tableName = tableConfig.getSourceTableName();
        String primaryKey = tableConfig.getPrimaryKey();
        String syncKey = getSyncKey(tableConfig);
        
        Long logId = null;
        int totalSynced = 0;
        
        // 🔥 关键优化：记录全量同步的开始时间作为基准点
        LocalDateTime fullSyncStartTime = LocalDateTime.now();
        long startTimestamp = System.currentTimeMillis();

        try {
            log.info("⏳ 开始全量同步 {} -> {}，基准时间：{}", 
                tableName, tableConfig.getTargetTableName(), fullSyncStartTime);
            
            // 创建日志记录
            if (syncLogService != null) {
                Long configId = getConfigId(tableConfig);
                logId = syncLogService.createLog(configId, tableName, tableConfig.getTargetTableName(), "FULL").getId();
            }

            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            log.info("🔄 全量同步开始 | 源表: {} -> 目标表: {}", tableName, tableConfig.getTargetTableName());
            
            // 步骤1：清空 TiDB 目标表
            log.info("🗑️  清空目标表...");
            dataSyncHandler.truncateTable(tableConfig);
            log.info("✅ 目标表已清空");

            // 步骤2：查询 StarRocks 全量数据（使用游标分页，性能优化）
            int batchSize = tableConfig.getBatchSize();
            
            // 🔥 智能选择分页策略：
            // 策略1：主键游标分页（推荐）- 适用于数字类型主键（INT, BIGINT等）
            // 策略2：时间戳游标分页（兜底）- 适用于字符串主键（UUID, VARCHAR等）
            String orderByField;
            String cursorField;
            String paginationStrategy;
            
            // 检测主键类型，决定使用哪种分页策略
            String primaryKeyType = detectPrimaryKeyType(tableName, primaryKey);
            if (isNumericType(primaryKeyType)) {
                // 策略1：主键是数字类型 -> 使用主键分页（性能最优）
                orderByField = primaryKey;
                cursorField = primaryKey;
                paginationStrategy = "PRIMARY_KEY";
                log.info("✅ 主键 {} 是数字类型({}), 使用【主键游标分页】策略", primaryKey, primaryKeyType);
            } else {
                // 策略2：主键是字符串或其他类型 -> 使用 update_time 分页（避免字符串比较性能问题）
                orderByField = tableConfig.getUpdateTimeField();
                cursorField = tableConfig.getUpdateTimeField();
                paginationStrategy = "UPDATE_TIME";
                log.info("⚠️ 主键 {} 是非数字类型({}), 使用【时间戳游标分页】策略", primaryKey, primaryKeyType);
            }
            
            Object lastCursorValue = null;      // 游标：上一批的最后一个值
            int batchCount = 0;
            
            log.info("📊 分页策略: {}, 排序字段: {}, 批大小: {}", paginationStrategy, orderByField, batchSize);

            // 🔥 注意：全量同步总是查询所有数据（包括软删除的）
            // 避免同一源表多次查询导致的数据不一致，软删除处理在目标表层面进行
            String softDeleteFilter = "";
            log.info("🚀 开始分批同步数据...");

            while (true) {
                long batchStartTime = System.currentTimeMillis();
                
                // 🔥 性能优化：使用 WHERE 字段 > ? 方式分页，避免 OFFSET
                // 即使ID无序（如UUID），此方式仍比 OFFSET 快100倍以上
                String sql;
                List<Map<String, Object>> batchData;
                
                if (lastCursorValue == null) {
                    // 第一批数据
                    sql = String.format(
                        "SELECT * FROM %s WHERE 1=1 %s ORDER BY %s ASC LIMIT ?",
                        tableName, softDeleteFilter, orderByField
                    );
                    batchData = starrocksJdbcTemplate.queryForList(sql, batchSize);
                } else {
                    // 后续批次：使用游标分页
                    sql = String.format(
                        "SELECT * FROM %s WHERE %s > ? %s ORDER BY %s ASC LIMIT ?",
                        tableName, cursorField, softDeleteFilter, orderByField
                    );
                    batchData = starrocksJdbcTemplate.queryForList(sql, lastCursorValue, batchSize);
                }

                if (batchData.isEmpty()) {
                    break; // 没有更多数据了
                }

                // 记录本批最后一个游标值，作为下一批的起点
                Map<String, Object> lastRow = batchData.get(batchData.size() - 1);
                lastCursorValue = lastRow.get(cursorField);

                // 同步这批数据
                long insertStartTime = System.currentTimeMillis();
                dataSyncHandler.batchInsert(tableConfig, batchData);
                long insertEndTime = System.currentTimeMillis();
                
                totalSynced += batchData.size();
                batchCount++;

                long batchEndTime = System.currentTimeMillis();
                long queryTime = insertStartTime - batchStartTime;
                long insertTime = insertEndTime - insertStartTime;
                long batchTime = batchEndTime - batchStartTime;
                
                log.info("📦 第{}批 | 本批: {} 条, 累计: {} 条 | 耗时[查询:{}ms, 插入:{}ms, 总计:{}ms] | 速度: {} 条/秒", 
                    batchCount, batchData.size(), totalSynced,
                    queryTime, insertTime, batchTime,
                    (int)(batchData.size() * 1000.0 / batchTime));
            }

            // 步骤3：使用开始时间作为同步基准点（而不是数据中的最大时间）
            // 这样可以确保全量同步期间的增量变更不会丢失
            lastSyncTimeMap.put(syncKey, fullSyncStartTime);
            
            long fullSyncEndTime = System.currentTimeMillis();
            long fullSyncDuration = fullSyncEndTime - startTimestamp;
            double avgSpeed = totalSynced * 1000.0 / fullSyncDuration;
            
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            log.info("✅ 全量同步完成 | 同步记录数: {}, 耗时: {} ms, 平均速度: {} 条/秒", 
                totalSynced, fullSyncDuration, (int)avgSpeed);
            log.info("📍 同步时间点设置为: {}", fullSyncStartTime);
            
            // 步骤4：🔥 立即执行一次增量补偿同步
            // 同步从全量开始到现在期间的所有变更数据
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            log.info("🔄 开始增量补偿同步（补偿全量期间的变更）...");
            int incrementalCount = syncIncrementalData(tableConfig, fullSyncStartTime);
            
            if (incrementalCount > 0) {
                log.info("✅ 增量补偿完成 | 补偿记录数: {}", incrementalCount);
                totalSynced += incrementalCount;
            }
            
            // 计算总体耗时
            long totalEndTime = System.currentTimeMillis();
            long totalDuration = totalEndTime - startTimestamp;
            double totalAvgSpeed = totalSynced * 1000.0 / totalDuration;
            
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            log.info("🎉 全量同步完成 | 总记录数: {}, 总耗时: {} ms, 总平均速度: {} 条/秒", 
                totalSynced, totalDuration, (int)totalAvgSpeed);
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            
            // 更新日志为成功
            if (syncLogService != null && logId != null) {
                LocalDateTime finalSyncTime = lastSyncTimeMap.get(syncKey);
                syncLogService.updateLog(logId, "SUCCESS", totalSynced, totalSynced, 0, finalSyncTime, null);
            }

        } catch (Exception e) {
            log.error("❌ 全量同步异常: {}", e.getMessage(), e);
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            
            // 更新日志为失败
            if (syncLogService != null && logId != null) {
                // 使用 fullSyncStartTime 而不是 maxUpdateTime，避免 LocalDateTime.MIN 导致数据库错误
                LocalDateTime safeLastSyncTime = fullSyncStartTime;
                syncLogService.updateLog(logId, "FAILED", totalSynced, totalSynced, 0, safeLastSyncTime, e.getMessage());
            }
            
            // 失败时也设置时间点，避免影响后续增量同步
            lastSyncTimeMap.put(syncKey, LocalDateTime.now());
        } finally {
            // 无论成功还是失败，都要清除全量同步标记
            fullSyncingMap.remove(syncKey);
            log.info("✅ 已清除全量同步标记，恢复增量同步");
        }
    }
    
    /**
     * 获取配置ID（从数据库配置中获取）
     */
    private Long getConfigId(SyncTableConfig tableConfig) {
        if (syncConfigService == null) {
            return null;
        }
        try {
            // 通过源表名和目标表名查找配置ID
            List<SyncConfigEntity> configs = syncConfigService.list();
            for (SyncConfigEntity config : configs) {
                if (config.getSourceTableName().equals(tableConfig.getSourceTableName())) {
                    return config.getId();
                }
            }
        } catch (Exception e) {
            log.warn("获取配置ID失败", e);
        }
        return null;
    }

    /**
     * 增量补偿同步（用于全量同步后补偿期间的变更数据）
     * @param tableConfig 表配置
     * @param startTime 开始时间
     * @return 同步的数据条数
     */
    private int syncIncrementalData(SyncTableConfig tableConfig, LocalDateTime startTime) {
        String tableName = tableConfig.getSourceTableName();
        String updateTimeField = tableConfig.getUpdateTimeField();
        String syncKey = getSyncKey(tableConfig);
        
        try {
            // 查询从 startTime 到现在的所有变更数据
            // 🔥 注意：总是查询所有数据（包括软删除的），软删除过滤在 handleUpdate 中处理
            String sql = String.format(
                "SELECT * FROM %s WHERE %s > ? ORDER BY %s ASC",
                tableName, updateTimeField, updateTimeField
            );
            
            String startTimeStr = startTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            log.info("🔄 增量补偿同步 | 源表: {} -> 目标表: {}", tableName, tableConfig.getTargetTableName());
            log.info("🔍 执行SQL: {}", sql);
            log.info("📌 参数: startTime={}", startTimeStr);
            
            List<Map<String, Object>> incrementalData = starrocksJdbcTemplate.queryForList(sql, startTimeStr);
            
            log.info("✅ 查询结果: {} 条数据", incrementalData.size());
            
            if (incrementalData.isEmpty()) {
                log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
                return 0;
            }
            
            log.info("🚀 开始补偿同步...");
            
            // 处理每条数据
            LocalDateTime maxUpdateTime = startTime;
            int syncCount = 0;
            
            for (Map<String, Object> rowData : incrementalData) {
                try {
                    dataSyncHandler.handleUpdate(tableConfig, rowData);
                    syncCount++;
                    
                    // 更新最大时间戳
                    Object updateTimeValue = rowData.get(updateTimeField);
                    LocalDateTime rowUpdateTime = convertToLocalDateTime(updateTimeValue, updateTimeField);
                    if (rowUpdateTime != null && rowUpdateTime.isAfter(maxUpdateTime)) {
                        maxUpdateTime = rowUpdateTime;
                    }
                } catch (Exception e) {
                    log.error("同步单条增量补偿数据失败", e);
                }
            }
            
            // 更新同步时间点
            lastSyncTimeMap.put(syncKey, maxUpdateTime);
            
            log.info("✅ 补偿同步完成 | 同步记录数: {}", syncCount);
            log.info("📍 更新同步时间点: {}", maxUpdateTime);
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            
            return syncCount;
            
        } catch (Exception e) {
            log.error("❌ 增量补偿同步异常: {}", e.getMessage(), e);
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            return 0;
        }
    }
    
    /**
     * 同步单个表的数据
     */
    private void syncTableData(SyncTableConfig tableConfig) {
        String tableName = tableConfig.getSourceTableName();
        String updateTimeField = tableConfig.getUpdateTimeField();
        String syncKey = getSyncKey(tableConfig);

        if (updateTimeField == null || updateTimeField.isEmpty()) {
            log.warn("表 {} -> {} 未配置update_time字段，跳过同步", 
                tableName, tableConfig.getTargetTableName());
            return;
        }

        // 🔥🔥🔥 第一次检查：在任何操作之前检查是否正在全量同步
        Boolean isFullSyncing = fullSyncingMap.get(syncKey);
        if (Boolean.TRUE.equals(isFullSyncing)) {
            // 全量同步中，静默跳过增量同步
            return;
        }

        // 获取上次同步时间
        LocalDateTime lastSyncTime = lastSyncTimeMap.get(syncKey);
        if (lastSyncTime == null) {
            lastSyncTime = LocalDateTime.now().minusMinutes(5); // 默认同步最近5分钟的数据
            log.warn("⚠️ 表 {} -> {} 的 lastSyncTime 为 null，使用默认值: {}", 
                tableName, tableConfig.getTargetTableName(), lastSyncTime);
        }
        
        Long logId = null;
        int successCount = 0;
        int failCount = 0;
        LocalDateTime maxUpdateTime = lastSyncTime;

        try {
            // 🔥🔥🔥 第二次检查：查询数据库前再次确认
            if (Boolean.TRUE.equals(fullSyncingMap.get(syncKey))) {
                return; // 静默跳过
            }

            // 查询增量数据
            // 🔥 注意：总是查询所有数据（包括软删除的），避免同一源表多次查询导致的数据不一致
            // 软删除的过滤逻辑在 handleUpdate 中根据目标表的 useSoftDelete 配置处理
            String sql = String.format(
                "SELECT * FROM %s WHERE %s > ? ORDER BY %s ASC LIMIT ?",
                tableName, updateTimeField, updateTimeField
            );

            List<Map<String, Object>> incrementalData = starrocksJdbcTemplate.queryForList(
                sql,
                Timestamp.valueOf(lastSyncTime),
                tableConfig.getBatchSize()
            );

            // 没有数据时静默跳过
            if (incrementalData.isEmpty()) {
                return;
            }
            
            // 有数据时才输出日志
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            log.info("📊 增量同步查询 | 源表: {} -> 目标表: {}", tableName, tableConfig.getTargetTableName());
            log.info("🔍 执行SQL: {}", sql);
            log.info("📌 参数: lastSyncTime={}, batchSize={}", lastSyncTime, tableConfig.getBatchSize());
            log.info("✅ 查询结果: {} 条数据", incrementalData.size());

            // 🔥🔥🔥 第三次检查：检测到数据后，同步前再次确认
            if (Boolean.TRUE.equals(fullSyncingMap.get(syncKey))) {
                log.warn("⚠️ 检测到全量同步正在进行，跳过本次增量同步");
                log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
                return;
            }

            log.info("🚀 开始同步数据...");
            
            // 创建日志记录（只在确认要同步时创建）
            if (syncLogService != null) {
                Long configId = getConfigId(tableConfig);
                logId = syncLogService.createLog(configId, tableName, tableConfig.getTargetTableName(), "INCREMENTAL").getId();
            }

            // 同步数据
            List<Map<String, Object>> failedDataList = new ArrayList<>();  // 收集失败的源数据
            
            for (Map<String, Object> rowData : incrementalData) {
                try {
                    // 尝试 UPSERT（先更新，更新失败则插入）
                    dataSyncHandler.handleUpdate(tableConfig, rowData);
                    successCount++;

                    // ⭐ 找出最晚的时间作为下次同步的起点
                    Object updateTimeValue = rowData.get(updateTimeField);
                    log.debug("字段 {} 的值：{}，类型：{}", updateTimeField, updateTimeValue, 
                        updateTimeValue != null ? updateTimeValue.getClass().getName() : "null");
                    LocalDateTime rowUpdateTime = convertToLocalDateTime(updateTimeValue, updateTimeField);
                    if (rowUpdateTime != null && rowUpdateTime.isAfter(maxUpdateTime)) {
                        log.debug("更新同步时间点：{} -> {}", maxUpdateTime, rowUpdateTime);
                        maxUpdateTime = rowUpdateTime;
                    }

                } catch (Exception e) {
                    log.error("同步 {} -> {} 的单行数据失败：{}", 
                        tableName, tableConfig.getTargetTableName(), rowData, e);
                    failCount++;
                    
                    // 收集失败的源数据（最多保存前10条）
                    if (failedDataList.size() < 10) {
                        failedDataList.add(rowData);
                    }
                }
            }

            // ⭐ 更新最后同步时间为最晚的数据时间
            lastSyncTimeMap.put(syncKey, maxUpdateTime);

            log.info("✅ 同步完成 | 成功: {}, 失败: {}", successCount, failCount);
            log.info("📍 更新同步时间点: {}", maxUpdateTime);
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            
            // 更新日志
            if (syncLogService != null && logId != null) {
                String status = failCount > 0 ? "FAILED" : "SUCCESS";
                String failedSourceData = null;
                
                // 如果有失败记录，将源数据转为JSON保存（仅保存前10条）
                if (failCount > 0 && !failedDataList.isEmpty()) {
                    try {
                        failedSourceData = JSON.toJSONString(failedDataList);
                    } catch (Exception e) {
                        log.warn("转换失败源数据为JSON失败", e);
                    }
                }
                
                syncLogService.updateLog(logId, status, successCount + failCount, successCount, failCount, 
                    maxUpdateTime, null, failedSourceData);
            }

        } catch (Exception e) {
            log.error("❌ 增量同步异常: {}", e.getMessage(), e);
            log.info("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
            
            // 更新日志为失败
            if (syncLogService != null && logId != null) {
                syncLogService.updateLog(logId, "FAILED", successCount + failCount, successCount, failCount, maxUpdateTime, e.getMessage());
            }
        }
    }

    /**
     * 应用启动时自动启动
     */
    @PostConstruct
    public void init() {
        if (syncConfig.getEnabled() && syncConfig.getAutoStart()) {
            log.info("配置了自动启动，延迟3秒后启动增量同步...");
            // 使用线程池异步启动
            fullSyncExecutor.submit(() -> {
                try {
                    Thread.sleep(3000); // 延迟3秒确保应用完全启动
                    startSync();
                } catch (Exception e) {
                    log.error("自动启动增量同步失败", e);
                }
            });
        }
    }

    /**
     * 应用关闭时自动停止
     */
    @PreDestroy
    public void destroy() {
        if (running) {
            stopSync();
        }
    }

    /**
     * 获取同步状态
     */
    public Map<String, Object> getSyncStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("running", running);
        status.put("mode", "INCREMENTAL_TIMESTAMP");
        
        // 使用实际生效的配置列表
        status.put("tableCount", activeTableConfigs.size());
        
        // 判断配置来源
        String configSource = "NONE";
        if (syncConfigService != null) {
            try {
                List<SyncConfigEntity> dbConfigs = syncConfigService.listEnabledConfigs();
                if (dbConfigs != null && !dbConfigs.isEmpty()) {
                    configSource = "DATABASE";
                }
            } catch (Exception e) {
                log.warn("检查数据库配置失败", e);
            }
        }
        if ("NONE".equals(configSource) && syncConfig != null && syncConfig.getEnabled() 
            && syncConfig.getTables() != null && !syncConfig.getTables().isEmpty()) {
            configSource = "YAML";
        }
        status.put("configSource", configSource);
        status.put("enabled", !activeTableConfigs.isEmpty());
        
        // 添加每个表的同步状态
        Map<String, Object> tableStatus = new HashMap<>();
        for (SyncTableConfig tableConfig : activeTableConfigs) {
            if (tableConfig.getEnabled()) {
                String syncKey = getSyncKey(tableConfig);
                Map<String, Object> info = new HashMap<>();
                info.put("lastSyncTime", lastSyncTimeMap.get(syncKey));
                info.put("updateTimeField", tableConfig.getUpdateTimeField());
                info.put("sourceTable", tableConfig.getSourceTableName());
                info.put("targetTable", tableConfig.getTargetTableName());
                info.put("isFullSyncing", Boolean.TRUE.equals(fullSyncingMap.get(syncKey))); // 是否正在全量同步
                tableStatus.put(syncKey, info);
            }
        }
        status.put("tables", tableStatus);
        
        // 添加全量同步统计信息
        long fullSyncingCount = fullSyncingMap.values().stream().filter(Boolean::booleanValue).count();
        status.put("fullSyncingCount", fullSyncingCount);
        
        return status;
    }

    /**
     * 是否正在运行
     */
    public boolean isRunning() {
        return running;
    }

    /**
     * 重置某个表的同步时间点（手动触发全量同步）
     */
    public void resetSyncTime(String tableName, LocalDateTime syncTime) {
        lastSyncTimeMap.put(tableName, syncTime);
        log.info("表 {} 的同步时间点已重置为：{}", tableName, syncTime);
    }

    /**
     * 将各种时间类型转换为 LocalDateTime
     * 支持：Timestamp, java.sql.Date, java.util.Date, LocalDateTime, String
     */
    private LocalDateTime convertToLocalDateTime(Object value, String fieldName) {
        if (value == null) {
            log.warn("字段 {} 的值为 null", fieldName);
            return null;
        }

        try {
            // 1. 已经是 LocalDateTime
            if (value instanceof LocalDateTime) {
                return (LocalDateTime) value;
            }

            // 2. Timestamp 类型
            if (value instanceof Timestamp) {
                return ((Timestamp) value).toLocalDateTime();
            }

            // 3. java.sql.Date 类型（只包含日期，补充时间为00:00:00）
            if (value instanceof java.sql.Date) {
                java.sql.Date sqlDate = (java.sql.Date) value;
                return sqlDate.toLocalDate().atStartOfDay();
            }

            // 4. java.util.Date 类型
            if (value instanceof java.util.Date) {
                return new Timestamp(((java.util.Date) value).getTime()).toLocalDateTime();
            }

            // 5. String 类型（尝试解析）
            if (value instanceof String) {
                String strValue = (String) value;
                // 尝试解析常见格式
                if (strValue.contains("T")) {
                    return LocalDateTime.parse(strValue);
                } else if (strValue.contains(" ")) {
                    return LocalDateTime.parse(strValue.replace(" ", "T"));
                }
            }

            log.warn("字段 {} 的值类型 {} 无法转换为 LocalDateTime，值：{}", 
                fieldName, value.getClass().getName(), value);
            return null;

        } catch (Exception e) {
            log.error("转换字段 {} 的值 {} 为 LocalDateTime 失败", fieldName, value, e);
            return null;
        }
    }
    
    /**
     * 检测主键的数据类型
     * @param tableName 表名
     * @param primaryKey 主键字段名
     * @return 数据类型（如：INT, BIGINT, VARCHAR等）
     */
    private String detectPrimaryKeyType(String tableName, String primaryKey) {
        try {
            String sql = String.format("SHOW FULL COLUMNS FROM %s WHERE Field = ?", tableName);
            List<Map<String, Object>> columns = starrocksJdbcTemplate.queryForList(sql, primaryKey);
            
            if (!columns.isEmpty()) {
                Map<String, Object> column = columns.get(0);
                String type = (String) column.get("Type");
                
                if (type != null) {
                    // 提取数据类型（去除长度限制）
                    // 例如：VARCHAR(50) -> VARCHAR, INT(11) -> INT, BIGINT UNSIGNED -> BIGINT
                    type = type.toUpperCase();
                    if (type.contains("(")) {
                        type = type.substring(0, type.indexOf("("));
                    }
                    if (type.contains(" ")) {
                        type = type.substring(0, type.indexOf(" "));
                    }
                    log.debug("检测到主键 {} 的类型为: {}", primaryKey, type);
                    return type;
                }
            }
            
            log.warn("无法检测主键 {} 的类型，默认为 VARCHAR", primaryKey);
            return "VARCHAR";
            
        } catch (Exception e) {
            log.error("检测主键类型失败，默认为 VARCHAR", e);
            return "VARCHAR";
        }
    }
    
    /**
     * 判断数据类型是否为数字类型
     * @param dataType 数据类型
     * @return true-数字类型，false-其他类型
     */
    private boolean isNumericType(String dataType) {
        if (dataType == null) {
            return false;
        }
        
        // 常见数字类型
        String type = dataType.toUpperCase();
        return type.equals("TINYINT") 
            || type.equals("SMALLINT") 
            || type.equals("INT") 
            || type.equals("INTEGER")
            || type.equals("BIGINT")
            || type.equals("MEDIUMINT")
            || type.equals("DECIMAL")
            || type.equals("NUMERIC")
            || type.equals("FLOAT")
            || type.equals("DOUBLE")
            || type.startsWith("INT") // INT UNSIGNED
            || type.startsWith("BIGINT"); // BIGINT UNSIGNED
    }
}

